package com.wk.springboot.bean;

import com.alibaba.fastjson.JSONObject;
import com.wk.springboot.adapter.ISearchHitAdapter;
import com.wk.springboot.adapter.SearchHitV561Adapter;
import com.wk.springboot.contants.Consts;
import com.wk.springboot.contants.ExportCont;
import com.wk.springboot.util.EsClientUtils;
import com.wk.springboot.util.EsExportUtils;
import com.wk.springboot.writer.MyFileWriter;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import org.apache.commons.lang.StringUtils;
import org.elasticsearch.action.search.SearchAction;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.NamedXContentRegistry;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.QueryParseContext;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.SearchModule;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.annotation.Configuration;


/**
 * @Classname EsExportBean
 * @Description es导出
 * @Date 2020/5/30 9:31
 * @Author by weikai
 */
//@Configuration
public class EsExportBean implements InitializingBean {

  private final Logger LOG = LoggerFactory.getLogger(EsExportBean.class);


  private ExportEntity exportEntity;

  private String clusterName;

  private String ip;

  private int port = 9300;

  private String user;

  private String index;

  private String type;

  private String jsonPath;

  private Map<String, ExportEntity.field> fields;

  private boolean trans;

  public static final String ONETOONE = "one2one";


  /**
   * 来自 ExportEntity 的字符串resultDefine, 或者为空时 来自 queryJSON
   * 在 startExport 开始调用时初始化
   */
  private ResultDefine resultDefine;

  /**
   * 通过指定的QSL查询json,从ES库中查询指定索引和类型中的数据,导出具有一对一的扁平结构的数据
   *
   * @throws IOException
   */

  public void startExport(String writeFilePath, String queryJSON, String esDataType, boolean append, String exportType) throws Exception{
    /* 可以解决 netty 的连接报错导致ES无法连接的问题(availableProcessors .... 之类),  */
    System.setProperty("es.set.netty.runtime.available.processors", "false");


    LOG.info("ElasticClient clusterName[{}] connecting...", clusterName);
    MyFileWriter writer = null;

    try{

      TransportClient client = EsClientUtils.createClient();
      // 创建写入对象
      writer = new MyFileWriter(writeFilePath, append);
      // 分批查询数据并写入文件
      SearchResponse responseOfData = getScrollID(queryJSON,client);
      // 查询第一个分页的ID
      SearchHits firstPageHit = responseOfData.getHits();
      //只导出为es的数据
      writeJsonToFile(firstPageHit,responseOfData,writer,client);

/*      if (ExportCont.Export.ELASTICSEARCH.equals(exportType)) {
        writeJsonToFile(firstPageHit,responseOfData,writer,client);
      } else {
        //导出类型为结构化数据库类型时
        writeDataTofile(firstPageHit,responseOfData,writer,append,queryJSON,esDataType,client);
      }*/
      // 将最后一批数据写入文件
      writer.close();
    } catch (Exception e) {
      throw new Exception("IOException ", e);
    }finally {
      if(writer != null){
        writer.close();
      }
    }
  }

 /* private void writeDataTofile(SearchHits firstPageHit, SearchResponse responseOfData, MyFileWriter writer, boolean append, String queryJSON, String esDataType, TransportClient client) throws Exception, InterruptedException {
    try {
      //导出类型为结构化数据库（非ELASTICSEARCH）时，将JSON数据转换为CSV格式的结构化数据
      // 将查询出的每行数据临时存储到List中
      List<String> lineList = new ArrayList<>();

      //此处去掉命令的插入，此部分操作放入导入时，进行
      //由于导出到ARTERYBASE数据库，需要在文件头添加语句
      //toABaseAppend(append, lineList);

      //version 5 将第一页文件写入文件
      //判断第一页是否为空
      if (firstPageHit.getHits().length != 0) {
        response2lineData(esDataType, writer, lineList, firstPageHit);
      }
      //应将下方循环放入第一页是否为空的判断中，考虑到尽量少的改造1.7.5 版本的代码故未放入
      //循环分页查找
      while (true) {
//        MonitorHelper.checkIn();
        responseOfData = client.prepareSearchScroll(responseOfData.getScrollId())
            .setScroll(new TimeValue(6000)).execute().actionGet();
        SearchHits searchHit = responseOfData.getHits();
        // 当查询不到数据时跳出循环
        if (searchHit.getHits().length == 0) {
          break;
        }
        response2lineData(esDataType, writer, lineList, searchHit);
      }
      writer.writeData(lineList);
    }catch (Exception e) {
      LOG.error("写入dat文件错误");
      throw new Exception("写入dat文件错误", e);
    }
  }*/

/*  private void response2lineData(String esDataType, MyFileWriter writer, List<String> lineList, SearchHits searchHit) throws Exception, InterruptedException {
    *//* resultDefine 若有效使用resultDefine 生成的fieldNames, 否则使用传入的 fieldNames  *//*
    String[] fieldNames = this.resultDefine.getFieldNames();
    // 遍历处理获取到的每个JSON文档/每行数据
    for (SearchHit hit : searchHit.getHits()) {
      SearchHitV561Adapter adapterHit = new SearchHitV561Adapter(hit);
      if (ONETOONE.equals(esDataType)) {
        String lineData = extractLineData(fieldNames, adapterHit);
        lineList.add(lineData);
      } else {
        List<String> nestedLineData = EsExportUtils.extractNestedLineData(fieldNames, adapterHit, fields, trans);
        lineList.addAll(nestedLineData);
      }

      // 分批次将数据写入数据文件
      if (lineList.size() % ExportCont.SIZE == 0) {
        writer.writeData(lineList);
        lineList.clear();
      }
    }
  }*/

  // 将查询出来的单个JOSN文档转换成一个文本行
/*  private String extractLineData(String[] fieldNames, ISearchHitAdapter hit) {
    // 将行信息存储到lineData中
    StringBuilder lineData = new StringBuilder();

    // 遍历处理获取到的每个JSON文档/每行数据

    // 遍历每个字段 取 fieldValue
    for (String fieldName : fieldNames) {
      String fieldValue = EsExportUtils.getFieldValueAssembly(fieldName, hit);
      *//* 获取fieldValue结束  *//*
      // 拼接字段信息
      lineData.append(fieldValue);
      lineData.append(Consts.COLUMN_SPLIT);
    }
    lineData.replace(lineData.length() - 1, lineData.length(), "");
    lineData.append("\r\n");
    return lineData.toString();
  }*/


  // 查询第一个分页的ID
  private SearchResponse getScrollID(String queryJSON, TransportClient client) throws Exception {

    //version 5
    SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
    SearchModule searchModule = new SearchModule(Settings.EMPTY, false, Collections.emptyList());

    //try()在()中内容执行后会自动关闭
    try (
        XContentParser parser = XContentFactory.xContent(XContentType.JSON)
            .createParser(new NamedXContentRegistry(searchModule.getNamedXContents()),queryJSON)
    ) {
      sourceBuilder.parseXContent(new QueryParseContext(parser));
    } catch (IOException e) {
      LOG.error("获取分页信息失败");
      throw new Exception("获取分页信息失败", e);
    }

    SearchRequestBuilder searchRequestBuilder = new SearchRequestBuilder(client, SearchAction.INSTANCE)
        .setSource(sourceBuilder)
        .setScroll(new TimeValue(6000))
        .setSearchType(SearchType.QUERY_THEN_FETCH)
        .setIndices(index);


    if (StringUtils.isNotBlank(type)) {
      searchRequestBuilder.setTypes(type);
    }

    // -----获取分页的响应信息
    return searchRequestBuilder.execute().actionGet();
  }

  private void writeJsonToFile(SearchHits firstPageHit, SearchResponse responseOfData, MyFileWriter writer, TransportClient client) throws Exception, InterruptedException {
    try {
      if (firstPageHit.getHits().length != 0) {
        extractLineData(firstPageHit,writer);
      }
      while (true) {
//        MonitorHelper.checkIn();
        responseOfData = client.prepareSearchScroll(responseOfData.getScrollId())
            .setScroll(new TimeValue(6000)).execute().actionGet();
        SearchHits searchHit = responseOfData.getHits();
        // 当查询不到数据时跳出循环
        if (searchHit.getHits().length == 0) {
          break;
        }
        extractLineData(searchHit,writer);
      }
    }catch (Exception e) {
      LOG.error("写入JSON文件错误");
      throw new Exception("写入JSON文件错误", e);
    }
  }

  private void extractLineData(SearchHits firstPageHit, MyFileWriter writer) throws Exception {
    for (SearchHit hit : firstPageHit) {
      String jsonLineData = JSONObject.parseObject(hit.getSourceAsString()).toString() + "\r\n";
      writer.writeData(jsonLineData);
    }
  }

  @Override
  public void afterPropertiesSet() throws Exception {

    this.index = "esyl";
    this.type = "yl";

    String filePath = "S:\\工作目录\\es\\yltest.dat";
    String query = "{    \"bool\": {\n"
        + "      \"filter\": {\n"
        + "        \"range\": {\n"
        + "          \"lrsj\": {\n"
        + "            \"gte\": \"2015-05-01\"\n"
        + "          }\n"
        + "        }\n"
        + "      }\n"
        + "    }}";
    startExport(filePath,query,"",true,"");
  }
}
